package com.hmy.rpc;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 1. 配置连接工厂<br>
 * 2. 建立TCP连接<br>
 * 3. 在TCP连接的基础上创建通道<br>
 * 4. 声明一个rpc_queue队列<br>
 * 5. 设置同时最多只能获取一个消息<br>
 * 6. 在rpc_queue队列在等待消息<br>
 * 7. 收到消息后，调用回调对象对消息进行处理，向此消息的replyTo队列中发送处理并带上correlationId<br>
 * 8. 使用wait-notify实现主线程和消息处理回调对象进行同步<br>
 * 
 * @author Administrator
 *
 */
public class RPCServer
{
	private static final String RPC_QUEUE_NAME = "rpc_queue";

	private static int fib(int n)
	{
		if (n == 0)
			return 0;
		if (n == 1)
			return 1;
		return fib(n - 1) + fib(n - 2);
	}

	public static void main(String[] args)
	{
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		factory.setPort(5682);

		try
		{
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();

			channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
			channel.basicQos(1);
			System.out.println(" [x] Awaiting RPC requests");

			// 定义消息的回调处理类，处理客户端发送的消息
			Consumer consumer = new DefaultConsumer(channel)
			{
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
						byte[] body) throws IOException
				{
					// 生产返回的结果配置信息，关键是设置correlationId值
					AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
							.correlationId(properties.getCorrelationId()).build();

					// 返回消息
					String response = "";

					try
					{
						String message = new String(body, "UTF-8");
						int n = Integer.parseInt(message);

						System.out.println(" [.] fib(" + message + ") " + properties.getCorrelationId());
						response += fib(n);
					} catch (RuntimeException e)
					{
						System.out.println(" [.] " + e.toString());
					} finally
					{
						// 回复消息，将服务端处理结果反馈给客户端
						channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
						// 对消息进行应答
						channel.basicAck(envelope.getDeliveryTag(), false);
						// RabbitMq consumer worker thread notifies the RPC
						// server owner thread
						// 唤起正在消费的所有线程
						synchronized (this)
						{
							this.notify();
						}
					}
				}
			};

			// 服务端接收客户端消息
			channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
			// Wait and be prepared to consume the message from RPC client.
			// 在接收到消息之前，服务端线程进入等待状态
			while (true)
			{
				synchronized (consumer)
				{
					try
					{
						consumer.wait();
					} catch (InterruptedException e)
					{
						e.printStackTrace();
					}
				}
			}

		} catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

}
